package spark.page;

import com.alibaba.fastjson.JSONObject;
import constant.Constants;
import dao.IPageSplitConvertRateDAO;
import dao.ITaskDAO;
import dao.factory.DAOFactory;
import domain.PageSplitConvertRate;
import domain.Task;
import org.apache.commons.collections.IteratorUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;
import test.MockData;
import util.DateUtils;
import util.NumberUtils;

import java.util.*;



public class PageStepConvertRateSpark {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName(Constants.SPARK_APP_NAME_PAGE).setMaster("local[2]");
        //jsc要在sparkSession前面
        JavaSparkContext jsc = new JavaSparkContext(conf);
        SparkSession sparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate();
        MockData.mockData(sparkSession);
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        Task task = taskDAO.findById(3);
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
        JavaRDD<Row> actionRDD = getActionRDDByDateRange(sparkSession, taskParam);
        JavaPairRDD<String, Row>  sessionid2actionRDD = getSessionid2actionRDD(actionRDD);
        sessionid2actionRDD = sessionid2actionRDD.cache();
        //累加器来完成其实页的统计
        LongAccumulator startPageAccumulator = new LongAccumulator();
        jsc.sc().register(startPageAccumulator, "startPageAccumulator");
        JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD = sessionid2actionRDD.groupByKey();

        JavaPairRDD<String,Integer> pageSplitRDD =  generateAndMatchPageSplit(jsc,sessionid2actionsRDD,taskParam,startPageAccumulator);
        Map<String, Long> pageSplitPVMap = pageSplitRDD.countByKey();
        //20
        Long startPagePv = startPageAccumulator.value();
        Map<String,Double> convertRateMap = computePageSplitConvertRate(taskParam,pageSplitPVMap,startPagePv);
        persistConvertRate(convertRateMap);






    }

    private static void persistConvertRate(Map<String, Double> convertRateMap) {
        StringBuffer stringBuffer = new StringBuffer();
        for (Map.Entry<String, Double> entry : convertRateMap.entrySet()) {
            String pageSplit = entry.getKey();
            Double converRate = entry.getValue();
            stringBuffer.append(pageSplit+"="+converRate+"|");
        }

        String converRate = stringBuffer.toString();
        PageSplitConvertRate pageSplitConvertRate = new PageSplitConvertRate();
        pageSplitConvertRate.setTaskid(new Date().getTime());
        pageSplitConvertRate.setConvertRate(converRate);
        IPageSplitConvertRateDAO pageSplitConvertRateDAO = DAOFactory.getPageSplitConvertRateDAO();
        pageSplitConvertRateDAO.insert(pageSplitConvertRate);


    }

    private static Map<String, Double> computePageSplitConvertRate(
            JSONObject taskParam,
            Map<String, Long> pageSplitPVMap,
            Long startPagePv) {
        HashMap<String, Double> convertRateMap = new HashMap<>();
        String[] targetPages = taskParam.getString(Constants.PARAM_TARGET_PAGE_FLOW).split(",");

        long prePageSplitPV = 0L;
        for (int i = 0; i < targetPages.length - 1; i++) {
            String targetPageSplit = targetPages[i] + "_" + targetPages[i + 1];
            Long targetPageSplitPv = pageSplitPVMap.get(targetPageSplit);

            double convertRate = 0.0;

            if(i==0){
                convertRate = NumberUtils.formatDouble((double) targetPageSplitPv/(double) startPagePv,2);
            }else {
                convertRate =  NumberUtils.formatDouble((double) targetPageSplitPv/(double) prePageSplitPV,2);
            }
            convertRateMap.put(targetPageSplit, convertRate);
            prePageSplitPV = targetPageSplitPv;
        }
        return convertRateMap;
    }

    private static JavaPairRDD<String, Integer> generateAndMatchPageSplit(
            JavaSparkContext jsc,
            JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD,
            JSONObject taskParam,
            LongAccumulator startPageAccumulator) {
        String taegetPageFlow = taskParam.getString(Constants.PARAM_TARGET_PAGE_FLOW);
        //广播变量；
        Broadcast<String> targetPageFlowBroadcast = jsc.broadcast(taegetPageFlow);


        JavaPairRDD<String, Integer> pageSplit2oneRDD = sessionid2actionsRDD.flatMapToPair(tuple -> {
            ArrayList<Tuple2<String, Integer>> list = new ArrayList<>();
            List<Row> rowList = IteratorUtils.toList(tuple._2.iterator());

            String[] targetPages = targetPageFlowBroadcast.value().split(",");
            String startpage = targetPages[0];
            Collections.sort(rowList, (row1, row2) -> {
                String actionTime1 = row1.getString(1);
                String actionTime2 = row2.getString(1);
                Date date1 = DateUtils.parseTime(actionTime1);
                Date date2 = DateUtils.parseTime(actionTime2);
                return (int) (date1.getTime() - date2.getTime());
            });
            //每次保留好前缀
            String prePageid = null;

            for (Row row : rowList) {
                String pageid = row.getString(4).substring(4);
                if(pageid.equals(startpage)){
                    startPageAccumulator.add(1L);
                }
                if (prePageid == null) {
                    prePageid = pageid;
                    continue;
                }
                String pageSplit = prePageid + "_" + pageid;
                for (int i = 0; i < targetPages.length - 1; i++) {
                    String targetPageSplit = targetPages[i] + "_" + targetPages[i + 1];
                    if (pageSplit.equals(targetPageSplit)) {
                        list.add(new Tuple2<>(pageSplit, 1));
                        break;
                    }
                }
                prePageid = pageid;
            }
            //flatMapToPair要返回的时迭代器记住；
            return list.iterator();
        });
        return pageSplit2oneRDD;
    }

    private static JavaPairRDD<String, Row>  getSessionid2actionRDD(JavaRDD<Row> actionRDD) {
        return actionRDD.mapToPair(row->{
            String sessionid = row.getString(3);
            return new Tuple2<>(sessionid, row);
        });
    }

    public static JavaRDD<Row> getActionRDDByDateRange(SparkSession sparkSession, JSONObject taskParam) {
        String startDate = taskParam.getString(Constants.PARAM_START_DATE);
        String endDate = taskParam.getString(Constants.PARAM_END_DATE);

        String sql = "select * " +
                "from user_visit_action " +
                "where date>='" + startDate + "' " +
                "and date<='" + endDate + "'";
        return sparkSession.sql(sql).toJavaRDD();

    }
}
